
/**
 * @name MASS L2 Heatwave Detects unusually hot days from simulation sample data
 * for January 2-31, 1980 Part of Pacific Northwest Climate Data Analysis
 * Project Proof of concept for MASS parallelization techniques to read netCDF
 * data
 *
 * @version 0.1.205 Alpha, MASS L2 Heatwave, 5th commit
 * @date 9/04/2013
 * @author Niko Simonson
 * @author Brett Yasutake
 *
 */
// IMPORTS
import MASS.*;                  // MASS Library
import java.io.IOException;     // IO Exceptions
import java.util.*;             // Vectors, etc.
import ucar.ma2.ArrayFloat;     // for netCDF reader
import ucar.ma2.IndexIterator;  // for netCDF reader
import ucar.nc2.NetcdfFile;     // for netCDF reader
import ucar.nc2.Variable;       // for netCDF reader

public class MassL2Heatwave extends Place {

    // CONSTANTS - MASS arguments
    private static final String[] MASSARGS = {"dslab", "dslab-302",
        "machinefile.txt", "12345"};    // connection arguments
    private static final int NPROCESSES = 1;    // number of processes to employ
    private static final int NTHREADS = 2;      // number of threads to employ
    private static final int HANDLE = 0;        // unique Places handle number
    // CONSTANTS - functions
    private static final int init_ = 0;          // initialize Places
    private static final int computeTemp_ = 1;   // compute regional avg temps
    private static final int maxTemp_ = 2;       // exchange regional temps
    private static final int computeMax_ = 3;    // compute daily max temps
    private static final int computeAvg_ = 4;
    private static final int reportMax_ = 5;
    // CONSTANTS - heatwave constants
    private static final int DAYS = 30;          // days of data collection
    private static final int DAYSEGMENTS = 4;    // time slices per day
    // target variable T2: temperature at 2 meters above surface
    private static final String TARGET = "T2";
    // northwest regional climate data sample
    private static final String FILENAMEBEGIN = "/home/brett/netcdf_files/1980_jan/wrfout_d02_1980-01-";
    private static final String FILENAMEEND = ":00:00";  // end of file name
    private static final int XRANGE = 123;
    private static final int YRANGE = 162;
    // VARIABLES
    private int sizeX, myX; // time slices in a day
    private int sizeY, myY; // days
    private double avgTemp; // regional average temperature ********************
    private String fileName; // target file
    private double[][] region;
    private static ProvenanceTracker logger;
    
//    private static Vector<double[]> finalTemps = new Vector<double[]>();
    
    private static ArrayList<double[]> finalTemps = new ArrayList<double[]>();

    // PARALLELIZATION FUNCTIONS
    /**
     * *************CONSTRUCTOR
     *
     ***********************************
     * @name MassL1Heatwave
     * @param interval unused Empty constructor required for MASS Place
     * implementation
     * ***********************************************************
     */
    public MassL2Heatwave(Object interval) {
        // A constructor implementation is required for MASS Place
        region = new double[XRANGE][YRANGE];
//        finalTemps = new Vector<double[]>();
    }

    /**
     * @name callMethod Provides enumerated list of functions to be parallelized
     * in MASS
     *
     * @param funcId - function ID (listed as Constants)
     * @param args - function parameters
     * @return - unused
     */
    public Object callMethod(int funcId, Object args) {
        switch (funcId) {
            case init_:
                return init(args);
            case computeTemp_:
                return computeTemp(args);
            case maxTemp_:
                return maxTemp(args);
            case computeMax_:
                return computeMax(args);
            case computeAvg_:
                return computeAvg(args);
            case reportMax_:
                return reportMax(args);
        }
        return null;
    }

    /**
     * @name init Initialization information for MASS Places index[0]: days
     * index[1]: time slices in a day
     *
     * @param args MASS initialization arguments, # of processes, # of threads
     * @return unused
     */
    public Object init(Object args) {
        // record maximum time ranges
        sizeX = DAYSEGMENTS;
        sizeY = DAYS;

        // Each place keeps track of its time position
        myX = index[1];
        myY = index[0];

        
        
        return null;
    }

    /**
     * @name computeTemp Open netCDF file and compute average temperature over
     * the entire region
     *
     * @param args unused
     * @return true = success, false = exception occurred
     */
    public Object computeTemp(Object args) {
        // VARIABLES
        NetcdfFile stubby = null; // target netCDF file

        avgTemp = 0.0;
        try {
            // open target netCDF file
            fileName = getFileName(myX, myY);
            stubby = NetcdfFile.open(fileName, null);	// Open the file

            // Find target variable
            Variable target = stubby.findVariable(TARGET);
            // Get list of file's variables
            List<Variable> stubbyVars = stubby.getVariables();
            Variable curr = stubbyVars.get(stubbyVars.indexOf(target));

            // Read values from variable (temperatures over the region)
            ArrayFloat.D3 d3values = (ArrayFloat.D3) curr.read();
            double totalTemp = 0.0;

            int x = 0;
            int y = 0;

            IndexIterator iter = d3values.getIndexIterator();
            while (iter.hasNext()) {
                iter.next();
//                totalTemp += (double) iter.getFloatCurrent();
                region[x][y] = (double) iter.getFloatCurrent();

                if (y == YRANGE - 1) {
                    x++;
                    y = 0;
                } else {
                    y++;
                }
            }
            // calculate regional average temperature
            avgTemp = totalTemp / d3values.getSize();
            return (Object) true;
        } catch (java.io.IOException e) {
            e.printStackTrace();
            return (Object) false;
        } catch (Exception e) {
            e.printStackTrace();
            return (Object) false;
        } finally {		// Close out the file no matter what occurs
            if (stubby != null) {
                try {
                    stubby.close();
                } catch (IOException ioe) {
                    ioe.printStackTrace();
                }
            }
//            Report regional average temperature
//            System.err.println("Day " + myY + " Hour " + (myX * 6) + " Temp: "
//                    + avgTemp);
            
            
            return (Object) null;
        }
    }

    /**
     * @name getFileName Private helper function to dynamically generate correct
     * file name
     *
     * @param thisX daily time slice
     * @param thisY day
     *
     * @return
     */
    private String getFileName(int thisX, int thisY) {
        // VARIABLES
        String strX, strY;  // file component numbers

        // calculate correct day of the month
        strY = Integer.toString(thisY + 2);

        // calculate correct time of the day
        strX = Integer.toString(thisX * 6);

        // adjust for numbers less than 10
        if (Integer.parseInt(strX) < 10) {
            strX = "0" + strX;
        }
        if (Integer.parseInt(strY) < 10) {
            strY = "0" + strY;
        }

        // return proper file name
        return FILENAMEBEGIN + strY + "_" + strX + FILENAMEEND;
    }

    /**
     * @name computeMax Computes maximum daily temperature
     * @param args unused
     * @return maximum daily temperature
     */
    public Object computeMax(Object args) {
        // VARIABLES
//        double max = avgTemp; // maximum temperature
        double[][] candidateRegion = new double[XRANGE][YRANGE];

        // Only midnight (00:00:00) of each day will do the work
        if (myX == 0) {
            // look at every other time slice for that day
            for (int i = 0; i < (DAYSEGMENTS - 1); ++i) {
                candidateRegion = (double[][]) inMessages[i];
                // if avg temp for that time slice is greater, use it
                for (int x = 0; x < XRANGE; ++x) {
                    for (int y = 0; y < YRANGE; ++y) {

                        if (region[x][y] < candidateRegion[x][y]) {
                            region[x][y] = candidateRegion[x][y];
                        }
                    }
                }
            } // next time slice
            return region;
        } // end if

        // return maximum average temperature for the day
        return null;
    }

    /**
     * @name maxTemp Used for ExchangeAll in preparation of finding maximum
     * temperature
     * @param args unused
     * @return each time slice's avg temp (max temp candidate)
     */
    public double[][] maxTemp(Object args) {
        return region;
    }

    public Object reportMax(Object args) {
        if (myX == 0) {
            return region;
        } else {
            return null;
        }
    }

    public Object computeAvg(Object args) {

        double[][] candidateRegion = new double[XRANGE][YRANGE];
        double[][] totals = new double[XRANGE][YRANGE];
        double[][] avgs = new double[XRANGE][YRANGE];
        double[][][] temps = new double[DAYS][XRANGE][YRANGE];
        double[][][] variances = new double[DAYS][XRANGE][YRANGE];
        double[][] localSDs = new double[XRANGE][YRANGE];
        double[][] thresholds = new double[XRANGE][YRANGE];
//        Vector<double[]> finalTemps = new Vector<double[]>();
        double[] record;

//        System.err.println("WE ARE HERE");



        if (myX == 0 && myY == 0) {

            //initialize arrays
            for (int x = 0; x < XRANGE; ++x) {
                for (int y = 0; y < YRANGE; ++y) {
                    totals[x][y] = region[x][y];
                    temps[0][x][y] = region[x][y];
                    localSDs[x][y] = 0.0;
                }
            }

            // adds all the future days temperatures by grid cell
            // temps stores the temperature values
            for (int i = 0; i < DAYS - 1; ++i) {
                candidateRegion = (double[][]) inMessages[i];

                for (int x = 0; x < XRANGE; ++x) {
                    for (int y = 0; y < YRANGE; ++y) {
                        totals[x][y] += candidateRegion[x][y];
                        temps[i + 1][x][y] = candidateRegion[x][y];
                    }
                }
            }

            // devide totals by DAYS to get average
            for (int x = 0; x < XRANGE; ++x) {
                for (int y = 0; y < YRANGE; ++y) {
                    avgs[x][y] = totals[x][y] / DAYS;
                    logger.log(1, x, y, -1, -1, avgs[x][y], false);
                }
            }

            // fill out variances with variances from average
            // add variances to localSDs
            for (int d = 0; d < DAYS; ++d) {
                for (int x = 0; x < XRANGE; ++x) {
                    for (int y = 0; y < YRANGE; ++y) {
                        variances[d][x][y] = Math.pow(temps[d][x][y] - avgs[x][y], 2.0);
//                        variances[d][x][y] = temps[d][x][y] - avgs[x][y];


                        localSDs[x][y] += variances[d][x][y];

                    }
                }
            }

            // calculate standard deviations in localSDs
            // calculate thresholds
            for (int x = 0; x < XRANGE; ++x) {
                for (int y = 0; y < YRANGE; ++y) {
                    localSDs[x][y] = Math.sqrt(localSDs[x][y] / (DAYS - 1));

                    thresholds[x][y] = avgs[x][y] + 2 * localSDs[x][y];
                    logger.log(2, x, y, -1, -1, thresholds[x][y], false);
                }
            }
            
            // find records that exceed threshold and store in finalTemps
            int count = 0;
            for (int d = 0; d < DAYS; ++d) {
                for (int x = 0; x < XRANGE; ++x) {
                    for (int y = 0; y < YRANGE; ++y) {

//                        
//                        if( d == x &&  x == y ) {
//                            System.err.println("DAY " + d + "-------------------");
//                            System.err.println("avgs["+ x +"][" + y + "] = " + avgs[x][y]);
//                            System.err.println("temps[" + d + "]["+ x +"][" + y + "] = " + temps[d][x][y]);
//                            System.err.println("localSDs["+ x +"][" + y + "] = " + localSDs[x][y]);
//                        }


                        //if the current (day,x,y) exceeds the threshold
                        if (temps[d][x][y] > thresholds[x][y]) {

                            record = new double[4];
                            
                            record[0] = d;
                            record[1] = x;
                            record[2] = y;
                            record[3] = temps[d][x][y];

//                          finalTemps.addElement(record);
                            finalTemps.add(record);
                            

//                            logger.log("*** computeMax record entry: " + count + ": ***\n" +
//                                       "                    live values -->  day: " + d + "   x: " + x + "   y: " + y + "   temperature: " + temps[d][x][y] + "\n" +
//                                       "                         record -->  day: " + record[0] + " x: " + record[1] + " y: " + record[2] + " temperature: " + record[3] + "\n" + 
//                                       "    after entry into finalTemps -->  day: " + finalTemps.get(count)[0] + " x: " + finalTemps.get(count)[1] + " y: " + finalTemps.get(count)[2] + " temperature: " + finalTemps.get(count)[3], true);

                            
//                            if (count > 0) {
//                                System.err.println("         ****PREVIOUS computeMax record entry: " + (count - 1) + ": ***");
////                                System.err.println("                    live values -->  day: " + d + "   x: " + x + "   y: " + y + "   temperature: " + temps[d][x][y]);
////                                System.err.println("                         record -->  day: " + record[0] + " x: " + record[1] + " y: " + record[2] + " temperature: " + record[3]);
//                                System.err.println("                               -->  day: " + finalTemps.get(count - 1)[0] + " x: " + finalTemps.get(count - 1)[1] + " y: " + finalTemps.get(count - 1)[2] + " temperature: " + finalTemps.get(count - 1)[3]);
//
//                            }
//                            
                            
                            ++count;
                        }
                    }

                }

            }
            
//            System.err.println("******************** finalTemps first entry day value: " + finalTemps.get(0)[0]);
            
            // debug
            if (finalTemps == null) {
//                logger.log("ALL NULL RESULTS", true);
            } else {
//                logger.log("NON NULL RESULTS", true);
//                logger.log("Final result set size before return in computeMax: " + finalTemps.size(), true);
//                for (int i = 0; i < finalTemps.size(); ++i) {
//                    System.err.println(finalTemps.elementAt(i)[0]);
//                }
            }

            return finalTemps;
        } else {
            return null;
        }
    }

    /**
     * @name main Driver to manage data analysis
     *
     * @param args unused
     * @throws Exception
     */
    public static void main(String[] args) throws Exception {
        // VARIABLES
        long start, end, totalTime; // data provenance for execution time
        // regional temperature calculations
        Object[] results = null;    // holds regional averages
        int count = DAYS * DAYSEGMENTS;     // total number of files
        // dummy object to prevent array over- or underflow
        Object[] fixer = new Object[count];
        Object[] fixer2 = new Object[count];
        double[] mxTemps = new double[DAYS]; // daily max temps
        // candidate max temps for one day
        double[] currRes = new double[DAYSEGMENTS];
        double maxTotal = 0.0; // total of all maximum temperatures 
        double[][] maxTotals = new double[XRANGE][YRANGE];
        // initialize maxTotals;
        double[][] localSD = new double[XRANGE][YRANGE];
        double[][] localThresholds = new double[XRANGE][YRANGE];
        for (int i = 0; i < XRANGE; ++i) {
            for (int j = 0; j < YRANGE; ++j) {
                maxTotals[i][j] = 0.0;
                localSD[i][j] = 0.0;
                localThresholds[i][j] = 0.0;
            }
        }
        int[] dests = new int[DAYS]; // partition of the start of each day
        Object[] mxResults = new Object[DAYS]; // maximum temps for each day
        // interim exchange all method
        Object[] mxResults2 = new Object[DAYS * DAYSEGMENTS];
        double[][] mxLocalResults = new double[XRANGE][YRANGE];
        double avgMax; // average daily high temperature
        // standard deviation calculations
        double[] deviations; // daily deviations from average high temp
        double[][][] localDeviations = new double[DAYS][XRANGE][YRANGE];
        double totalDev = 0.0; // total of daily deviations    
        double standardDev;  // standard deviation

        // hot day detection results calculations
        double hotDayThreshold; // temperature to count as a "hot day"
        // holds hot days
        HashMap<Integer, Double> hotDays = new HashMap<Integer, Double>();
        Set<Integer> heatwave;  // days that were hot
        Iterator<Integer> finalIter; // iterator over set of daily max temps
        
        logger = new ProvenanceTracker();

        
        Vector<double[]> report = new Vector<double[]>();
        Object[] fnl;
        
        // begin time measurement
        start = new Date().getTime();

        // initialize MASS
        MASS.init(MASSARGS, NPROCESSES, NTHREADS);

        // Create Places
        // Each computing node corresponds to one netCDF file
        Places computeHeatWave = new Places(HANDLE, "MassL2Heatwave", null,
                DAYS, DAYSEGMENTS);

        // Create exchangeAll vector, 
        // corresponding to the three time slices after midnight
        Vector<int[]> segTemps = new Vector<int[]>();
        int[] oneNextSeg = {0, 1};
        int[] twoNextSeg = {0, 2};
        int[] threeNextSeg = {0, 3};
        segTemps.add(oneNextSeg);
        segTemps.add(twoNextSeg);
        segTemps.add(threeNextSeg);

        int[][] daySegs = new int[DAYS][2];
        Vector<int[]> segLocals = new Vector<int[]>();
        for (int i = 1; i < DAYS; ++i) {
            daySegs[i][0] = i;
            daySegs[i][1] = 0;
            segLocals.add(daySegs[i]);
        }

        
        
//        logger.log("logger.log is writing this");


        // Initialize Places
        computeHeatWave.callAll(init_, null);

        // compute regional average temperature at each time slice
        results = computeHeatWave.callAll(computeTemp_, fixer);

        int counter = 0;
        // Calculation of standard deviation
        // get daily max temps
        for (int i = 0; i < DAYS; ++i) {
            dests[i] = i * DAYSEGMENTS;
        }
        try {
            // each file reports its regional average temperature
            computeHeatWave.exchangeAll(HANDLE, maxTemp_, segTemps);

            // dummy object to keep ExchangeAll from array over- or underflow
            Object[] someFix = new Object[3];

            // every day at midnight gets the correct max temp 
            // (e.g. 0, 4, 8, etc.)
            mxResults2 = computeHeatWave.callAll(computeMax_, fixer);


//            if (mxResults2 == null) {
//                System.err.println("It's null");
//            } else {
//                for (int i = 0; i < mxResults2.length; ++i) {
//                    System.err.println("maxedResult (" + (i + 1) + "): "
//                            + mxResults2[i]);
//                }
//            }

////////////////////////////////////////////////////////////////////////////////


            computeHeatWave.exchangeAll(HANDLE, reportMax_, segLocals);
            fnl = computeHeatWave.callAll(computeAvg_, fixer2);
            
   
//            logger.log("Final result set size from MASS controller: " + fnl.length, true);
//            Object getThis = fnl[0];
            Object getThis = null;
            
//            System.err.println(getThis);
            
            
            
            // get total execution time
            end = new Date().getTime();
            totalTime = end - start;    //milliseconds
            
            // get the only non-null result from call all computeAvg results
            int cntr = 0;
            while (cntr < fnl.length) {
                if (fnl[cntr] != null) {
                    getThis = fnl[cntr];
                }
                ++cntr;
            }
            //check that you received data from 0,0 node
//            if (getThis == null) {
////                logger.log(4"Final results returned from node (0,0) are null", true);
//            }
//            else {
////                logger.log("Node (0,0) returned non-null results", true);
//            }
//            Vector<double[]> getThis2 = (Vector<double[]>)getThis;
            
//            System.err.println(test.length);
            
//            System.err.println("finalTemps size after extracting from return value: " + getThis2.size());
//            
//            for (int i = 0; i < getThis2.size(); ++i) {
//                System.err.println("Day: " + getThis2.elementAt(i)[0] 
//                        + " X: " + getThis2.elementAt(i)[1]
//                        + " Y: " + getThis2.elementAt(i)[2]
//                        + " Daily Max Temp: " + getThis2.elementAt(i)[3]);
//            }
            
//            System.err.println("finalTemps size after extracting from return value: " + finalTemps.size());
//            
//            for (int i = 0; i < finalTemps.size(); ++i) {
//                System.err.println("Day: " + finalTemps.get(i)[0] 
//                        + " X: " + finalTemps.get(i)[1]
//                        + " Y: " + finalTemps.get(i)[2]
//                        + " Daily Max Temp: " + finalTemps.get(i)[3]);
//            }
            
              // report final results
              for (int i = 0; i < finalTemps.size(); ++i) {
                logger.log(3, (int)finalTemps.get(i)[1], (int) finalTemps.get(i)[2], (int) finalTemps.get(i)[0], -1, finalTemps.get(i)[3], false);
                        
                        
//                        "Day " + (int) finalTemps.get(i)[0]
//                        + ": (" + (int) finalTemps.get(i)[1]
//                        + ", " + (int) finalTemps.get(i)[2]
//                        + ") --> Daily Max Temp: " + finalTemps.get(i)[3], true);
            }
           

//            // initialize mxResults from mxResults2
//            for (int j = 0; j < DAYS; ++j) {
//                mxResults[j] = mxResults2[j * 4];
//                mxLocalResults = (double[][]) mxResults[j];
//                //maxTotal += (double) mxResults[j];
//                for (int x = 0; x < XRANGE; ++x) {
//                    for (int y = 0; y < YRANGE; ++y) {
//                        maxTotals[x][y] += mxLocalResults[x][y];
//                    }
//                }
//            }
//
//            computeHeatWave.callAll(computeAvg_, mxResults);
//
//
//            for (int x = 0; x < XRANGE; ++x) {
//                for (int y = 0; y < YRANGE; ++y) {
//                    maxTotals[x][y] /= DAYS;
//                }
//            }
//
//            for (int day = 0; day < DAYS; ++day) {
//                mxLocalResults = (double[][]) mxResults[day];
//                for (int x = 0; x < XRANGE; ++x) {
//                    for (int y = 0; y < YRANGE; ++y) {
//                        localDeviations[day][x][y] = Math.pow(mxLocalResults[x][y] - maxTotals[x][y], 2.0);
//                        localSD[x][y] += localDeviations[day][x][y];
//                    }
//                }
//            }
//            //System.err.println("THRESHOLDS");
//            for (int x = 0; x < XRANGE; ++x) {
//                for (int y = 0; y < YRANGE; ++y) {
//                    localSD[x][y] = Math.sqrt(localSD[x][y] / (DAYS - 1));
//                    localThresholds[x][y] = maxTotals[x][y] + (2 * localSD[x][y]);
//                    //System.err.println("X: " + x + ", Y: " + y + " Threshold: " + localThresholds[x][y]);
//                }
//            }
//
//            System.err.println("RESULTS");
//            for (int day = 0; day < DAYS; ++day) {
//                mxLocalResults = (double[][]) mxResults[day];
//                for (int x = 0; x < XRANGE; ++x) {
//                    for (int y = 0; y < YRANGE; ++y) {
//                        if (mxLocalResults[x][y] > localThresholds[x][y]) {
//                            System.err.println("Day: " + day + " X: " + x + ", y: " + y + " Temperature: " + mxLocalResults[x][y]);
//                        }
//                    }
//                }
//            }
//            // calculate the average temperature from the maximums
////            avgMax = maxTotal / DAYS;
//
//            // calculate the standard deviation
////            deviations = new double[mxResults.length];
//
////            for (int i = 0; i < mxResults.length; ++i) {
////                deviations[i] = Math.pow((Double) mxResults[i] - avgMax, 2.0);
////                totalDev += deviations[i];
////            }
////
////            standardDev = Math.sqrt(totalDev / (DAYS - 1));
//
////            hotDayThreshold = avgMax + (1 * standardDev);
//
////            for (int i = 0; i < DAYS; ++i) {
////                if ((Double) mxResults[i] > hotDayThreshold) {
////                    mxTemps[i] = (Double) mxResults[i];
////
////                    hotDays.put(i, mxTemps[i]);
////                }
////            }

            

//            System.err.println("Final Results (Days Hotter Than "
//                    + hotDayThreshold + "):");

//            heatwave = hotDays.keySet();
//            finalIter = heatwave.iterator();
//            while (finalIter.hasNext()) {
//                int day = finalIter.next();
//
//                // report final analysis results
//                System.err.println("Day: " + day + "  Temp: "
//                        + hotDays.get(day));
//            }

            // report data provenance (execution time)
//            logger.log("Total execution time (milliseconds): "
//                    + totalTime);
            
            System.err.println("Total execution time (milliseconds): "
                    + totalTime);

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // finish MASS
            MASS.finish();
        }
    }
} // end MASS Heatwave
